//
// Created by mi on 2024/1/28.
//
#include <iostream>
#include <sstream>
#include <string>
#include <librdkafka/rdkafkacpp.h>

//const std::string broker = "192.168.0.250:9092";
const std::string broker = "172.20.10.5:2181";
const std::string topic = "yezongwen_topic";

class ExampleDeliveryReportCb : public RdKafka::DeliveryReportCb {
public:
    void dr_cb(RdKafka::Message &message) {
        /* If protobuf_demo.err() is non-zero the protobuf_demo delivery failed permanently
         * for the protobuf_demo. */
        if (message.err()) {
            std::cerr << "% Message delivery failed: " << message.errstr()
                      << std::endl;
        } else {
            std::cerr << "% Message delivered to topic " << message.topic_name()
                      << " [" << message.partition() << "] at offset "
                      << message.offset() << std::endl;
        }
    }
};

int main() {
    std::shared_ptr<RdKafka::Conf> conf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
    std::string errstr;
    ExampleDeliveryReportCb deliveryReport;

    if (conf->set("bootstrap.servers", broker, errstr) !=
        RdKafka::Conf::CONF_OK) {
        std::cerr << errstr << std::endl;
        return 0;
    }

    if (conf->set("dr_cb", &deliveryReport, errstr) != RdKafka::Conf::CONF_OK) {
        std::cerr << errstr << std::endl;
        return 0;
    }

    std::shared_ptr<RdKafka::Producer> producer(RdKafka::Producer::create(conf.get(), errstr));
    if (!producer) {
        std::cerr << "Failed to create producer: " << errstr << std::endl;
        return 0;
    }
    std::string msg = "yezongwen";
    for (int i = 0; i < 2; i++) {
        RdKafka::ErrorCode err = producer->produce(
                /* Topic name */
                topic,
                /* Any Partition: the builtin partitioner will be
                 * used to assign the protobuf_demo to a topic based
                 * on the protobuf_demo key, or random partition if
                 * the key is not set. */
                RdKafka::Topic::PARTITION_UA,
                /* Make a copy of the value */
                RdKafka::Producer::RK_MSG_COPY /* Copy payload */,
                /* Value */
                const_cast<char *>(msg.c_str()), msg.size(),
                /* Key */
                nullptr, 0,
                /* Timestamp (defaults to current time) */
                0,
                /* Message headers, if any */
                nullptr,
                /* Per-protobuf_demo opaque value passed to
                 * delivery report */
                nullptr);

        if (err != RdKafka::ERR_NO_ERROR) {
            std::cerr << "% Failed to produce to topic " << topic << ": "
                      << RdKafka::err2str(err) << std::endl;
        }
        //这样就知道 消息发送回调。
        producer->poll(0);
    }
    producer->flush(10 * 1000 /* wait for max 10 seconds */);

    if (producer->outq_len() > 0) {
        std::cerr << "% " << producer->outq_len()
                  << " protobuf_demo(s) were not delivered" << std::endl;
    }
    return 0;
}


